package javaVersion.learn.checkPoint;

import javaVersion.pojo.SEvent;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.streaming.api.checkpoint.ListCheckpointed;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;

/**
 * @program: myFlink
 * @description: 带水印的checkpoint，窗口中的处理逻辑，带有快照
 * @author: WincoMa
 * @create: 2020-07-23 17:33
 **/
public class Eg1_WindowStatisticWithChk implements WindowFunction<SEvent, Long, Tuple, TimeWindow>, ListCheckpointed<Eg1_UDFState> {
    private long total = 0L;

    /**
     * 制作自定义状态快照
     * @param checkpointId 唯一单调递增的数字
     * @param timestamp  master触发checkpoint的时间戳
     * @return 一个保存各个状态的List集合
     */
    @Override
    public List<Eg1_UDFState> snapshotState(long checkpointId, long timestamp) {
        ArrayList<Eg1_UDFState> udfList = new ArrayList<>();
        //将想要代表状态的内容封装到状态类（Eg1_UDFState）中
        Eg1_UDFState udfState = new Eg1_UDFState();
        udfState.setState(total);
        //将封装好的状态放入一个列表中，并返回这个列表，作为快照
        udfList.add(udfState);
        return udfList;
    }

    /**
     * 从自定义快照中恢复状态
     * @param state
     * @throws Exception
     */
    @Override
    public void restoreState(List<Eg1_UDFState> state) throws Exception {
        Eg1_UDFState udfState = state.get(0);
        total = udfState.getState();
    }

    /**
     * window算子的实现逻辑，即:统计window中元组的数量
     * @param tuple
     * @param window
     * @param input
     * @param out
     * @throws Exception
     */
    @Override
    public void apply(Tuple tuple, TimeWindow window, Iterable<SEvent> input, Collector<Long> out) throws Exception {
        long count = 0L;
        //遍历进入每个窗口中的元素，
        // 在每个单独窗口中，每遍历一个元素，+1，统计窗口中元素数量
        for (SEvent event : input) {
            count += 1L;
        }
        total += count;
        out.collect(count);
    }
}
